/**
 * 
 */
package com.redis.spi;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.ShardedJedis;
import redis.clients.jedis.ShardedJedisPool;

import com.redis.api.DataWriter;
import com.redis.io.utils.HessianSerializerUtils;

/**
 * @author yangjuanying
 */
public class JedisDataWriter implements DataWriter {
	
	/**
	 * jedis只写客户端连接池
	 */
	private final ShardedJedisPool shardedJedisPool;
	
	public JedisDataWriter(final ShardedJedisPool writeOnlyPool){
		this.shardedJedisPool = writeOnlyPool;
	}
	
	/**
	 * 回收jedis客户端
	 * @param isNormal                 是否正常
	 * @param jedis                    jedis
	 */
	private void release(boolean isNormal, ShardedJedis jedis){
		if (isNormal)
			this.shardedJedisPool.returnResource(jedis);
		else {
			this.shardedJedisPool.returnBrokenResource(jedis);
		}
	}
	
	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#set(byte[], byte[])
	 */
	public boolean set(byte[] key, byte[] value) {
		return set(key, value, -1);
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#setnx(byte[], byte[])
	 */
	public boolean setnx(byte[] key, byte[] value) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long exRet = null;
		try {
			exRet = jedis.setnx(key, value);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return exRet == null ? false: (exRet.intValue()==1? true: false);
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#setnx(java.lang.String, java.lang.String)
	 */
	public boolean setnx(String key, String value) {
		return setnx(key, value, -1);
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#setnx(java.lang.String, java.lang.String, int)
	 */
	public boolean setnx(String key, String value, int expire) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long exRet = null;
		try {
			exRet = jedis.setnx(key, value);
			if(exRet != null && exRet.intValue() ==1){
				if(expire > 0){
					jedis.expire(key, expire);
					flag = true;
				}
			}
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return flag;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#set(byte[], byte[], int)
	 */
	public boolean set(byte[] key, byte[] value, int expire) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		String ret = null;
		try {
			if(expire > 0){
				ret = jedis.setex(key, expire, value);
			}else{
				ret = jedis.set(key, value);
			}
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return "OK".equals(ret) ? true : false;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#set(java.lang.String, java.lang.String)
	 */
	public boolean set(String key, String value) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		String ret = null;
		try {
			ret = jedis.set(key, value);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return "OK".equals(ret) ? true : false;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#set(java.lang.String, java.lang.String, int)
	 */
	public boolean set(String key, String value, int expire) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		String ret = null;
		try {
			ret = jedis.setex(key, expire, value);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return "OK".equals(ret) ? true : false;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#setBean(java.lang.String, java.io.Serializable)
	 */
	public <T extends Serializable> boolean setBean(String key, T value) {
		return setBean(key, value, -1);
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#setBean(java.lang.String, java.io.Serializable, int)
	 */
	public <T extends Serializable> boolean setBean(String key, T value,
			int expire) {
		return set(key.getBytes(), HessianSerializerUtils.serialize(value), expire);
	}
	
	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#getSet(byte[], byte[])
	 */
	public byte[] getSet(byte[] key, byte[] value) {
		return getSet(key, value, -1);
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#getSet(byte[], byte[], int)
	 */
	public byte[] getSet(byte[] key, byte[] value, int expire) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		byte[] ret = null;
		try {
			ret = jedis.getSet(key, value);
			if(expire > 0){
				jedis.expire(key, expire);
			}
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return ret;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#getSet(java.lang.String, java.lang.String)
	 */
	public String getSet(String key, String value) {
		return getSet(key, value, -1);
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#getSet(java.lang.String, java.lang.String, int)
	 */
	public String getSet(String key, String value, int expire) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		String ret = null;
		try {
			ret = jedis.getSet(key, value);
			if(expire > 0){
				jedis.expire(key, expire);
			}
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return ret;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#getSetBean(java.lang.String, java.io.Serializable)
	 */
	public <T extends Serializable> T getSetBean(String key, T value) {
		return getSetBean(key, value, -1);
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#getSetBean(java.lang.String, java.io.Serializable, int)
	 */
	@SuppressWarnings("unchecked")
	public <T extends Serializable> T getSetBean(String key, T value,
			int expire) {
		return (T)HessianSerializerUtils.deSerialize(getSet(key.getBytes(), HessianSerializerUtils.serialize(value), expire));
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#append(java.lang.String, byte[])
	 */
	public long append(String key, byte[] value) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long ret = null;
		try {
			ret = jedis.append(key.getBytes(), value);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return ret == null ? 0 : ret;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#append(java.lang.String, java.lang.String)
	 */
	public long append(String key, String value) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long ret = null;
		try {
			ret = jedis.append(key, value);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return ret == null ? 0 : ret;
	}
	
	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#mset(java.lang.String[])
	 */
	public boolean mset(String... kvs) {
		if(kvs != null && kvs.length > 0 && kvs.length%2 ==0){
			ShardedJedis jedis = this.shardedJedisPool.getResource();
			boolean flag = false;
			String ret = null;
			try {
				Jedis j = jedis.getShard(kvs[0]);
				ret = j.mset(kvs);
				flag = true;
			} catch (Exception e) {
				flag = false;
				e.printStackTrace();
			} finally {
				release(flag, jedis);
			}
			return "OK".equals(ret) ? true : false;
		}else{
			return false;
		}
	}
	
	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#msetnx(java.lang.String[])
	 */
	public long msetnx(String... kvs) {
		if(kvs != null && kvs.length > 0 && kvs.length%2 ==0){
			ShardedJedis jedis = this.shardedJedisPool.getResource();
			boolean flag = false;
			Long ret = null;
			try {
				Jedis j = jedis.getShard(kvs[0]);
				ret = j.msetnx(kvs);
				flag = true;
			} catch (Exception e) {
				flag = false;
				e.printStackTrace();
			} finally {
				release(flag, jedis);
			}
			return ret == null ? 0 : ret;
		}else{
			return 0;
		}
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#del(java.lang.String)
	 */
	public boolean del(String key) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		try {
			jedis.del(key);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return flag;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#incr(java.lang.String)
	 */
	public long incr(String key) {
		return incrBy(key, 1);
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#incrBy(java.lang.String, long)
	 */
	public long incrBy(String key, long step) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long ret = null;
		try {
			ret = jedis.incrBy(key, step);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return ret == null ? 0 : ret;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#decr(java.lang.String)
	 */
	public long decr(String key) {
		return decrBy(key, 1);
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#decrBy(java.lang.String, long)
	 */
	public long decrBy(String key, long step) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long ret = null;
		try {
			ret = jedis.decrBy(key, step);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return ret == null ? 0 : ret;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#incrByFloat(java.lang.String, double)
	 */
	public double incrByFloat(String key, double step) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Double ret = null;
		try {
			Jedis j = jedis.getShard(key);
			ret = j.incrByFloat(key, step);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return ret == null ? 0 : ret;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#hset(java.lang.String, java.lang.String, byte[])
	 */
	public boolean hset(String key, String field, byte[] value) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		try {
			jedis.hset(key.getBytes(), field.getBytes(), value);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return flag;
	}
	
	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#hsetnx(java.lang.String, java.lang.String, byte[])
	 */
	public boolean hsetnx(String key, String field, byte[] value) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		try {
			jedis.hsetnx(key.getBytes(), field.getBytes(), value);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return flag;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#hset(java.lang.String, java.lang.String, java.lang.String)
	 */
	public boolean hset(String key, String field, String value) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		try {
			jedis.hset(key, field, value);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return flag;
	}
	
	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#hmset(java.lang.String, java.util.Map)
	 */
	@Override
	public boolean hmset(String key, Map<String, String> fieldValues) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		if(fieldValues == null || fieldValues.isEmpty()){
			return flag;
		}
		try {
			String ret = jedis.hmset(key, fieldValues);
			if("OK".equals(ret)){
				flag = true;
			}else{
				System.out.println("hmset error:" + ret);
				flag = false;
			}
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return flag;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#hsetnx(java.lang.String, java.lang.String, java.lang.String)
	 */
	public boolean hsetnx(String key, String field, String value) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		try {
			jedis.hsetnx(key, field, value);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return flag;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#hsetBean(java.lang.String, java.lang.String, java.io.Serializable)
	 */
	public <T extends Serializable> boolean hsetBean(String key, String field,
			T value) {
		return hset(key, field, HessianSerializerUtils.serialize(value));
	}
	
	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#hsetnxBean(java.lang.String, java.lang.String, java.io.Serializable)
	 */
	public <T extends Serializable> boolean hsetnxBean(String key,
			String field, T value) {
		return hsetnx(key, field, HessianSerializerUtils.serialize(value));
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#hdel(java.lang.String, java.lang.String[])
	 */
	public long hdel(String key, String... fields) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long exRet = null;
		try {
			exRet = jedis.hdel(key, fields);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return exRet == null ? 0 : exRet;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#hincrBy(java.lang.String, java.lang.String, long)
	 */
	public long hincrBy(String key, String field, long step) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long exRet = null;
		try {
			exRet = jedis.hincrBy(key, field, step);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return exRet == null ? 0 : exRet;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#srem(java.lang.String, byte[][])
	 */
	public long srem(String key, byte[]... members) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.srem(key.getBytes(), members);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#srem(java.lang.String, java.lang.String[])
	 */
	public long srem(String key, String... members) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.srem(key, members);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}
	
	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#lpush(java.lang.String, byte[][])
	 */
	public long lpush(String key, byte[]... members) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.lpush(key.getBytes(), members);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#lpush(java.lang.String, java.lang.String[])
	 */
	@Override
	public long lpush(String key, String... members) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.lpush(key, members);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#lpush(java.lang.String, java.io.Serializable[])
	 */
	public <T extends Serializable> long lpush(String key, T... members) {
		if(members != null && members.length >0){
			byte[][] tmp = new byte[members.length][];
			for(int i=0;i<tmp.length;i++){
				tmp[i] = HessianSerializerUtils.serialize(members[i]);
			}
			return lpush(key, tmp);
		}
		return 0;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#lpushx(java.lang.String, byte[][])
	 */
	public long lpushx(String key, byte[]... members) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.lpushx(key.getBytes(), members);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#lpushx(java.lang.String, java.lang.String[])
	 */
	public long lpushx(String key, String... members) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.lpushx(key, members);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#lpushx(java.lang.String, java.io.Serializable[])
	 */
	public <T extends Serializable> long lpushx(String key, T... members) {
		if(members != null && members.length >0){
			byte[][] tmp = new byte[members.length][];
			for(int i=0;i<tmp.length;i++){
				tmp[i] = HessianSerializerUtils.serialize(members[i]);
			}
			return lpushx(key, tmp);
		}
		return 0;
	}
	
	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#rpush(java.lang.String, byte[][])
	 */
	public long rpush(String key, byte[]... members) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.rpush(key.getBytes(), members);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#rpush(java.lang.String, java.lang.String[])
	 */
	public long rpush(String key, String... members) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.rpush(key, members);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#rpush(java.lang.String, java.io.Serializable[])
	 */
	public <T extends Serializable> long rpush(String key, T... members) {
		if(members != null && members.length >0){
			byte[][] tmp = new byte[members.length][];
			for(int i=0;i<tmp.length;i++){
				tmp[i] = HessianSerializerUtils.serialize(members[i]);
			}
			return rpush(key, tmp);
		}
		return 0;
	}
	
	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#rpushx(java.lang.String, byte[][])
	 */
	public long rpushx(String key, byte[]... members) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.rpushx(key.getBytes(), members);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#rpushx(java.lang.String, java.lang.String[])
	 */
	public long rpushx(String key, String... members) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.rpushx(key, members);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#rpushx(java.lang.String, java.io.Serializable[])
	 */
	public <T extends Serializable> long rpushx(String key, T... members) {
		if(members != null && members.length >0){
			byte[][] tmp = new byte[members.length][];
			for(int i=0;i<tmp.length;i++){
				tmp[i] = HessianSerializerUtils.serialize(members[i]);
			}
			return rpushx(key, tmp);
		}
		return 0;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#sadd(java.lang.String, byte[][])
	 */
	public long sadd(String key, byte[]... members) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.sadd(key.getBytes(), members);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#sadd(java.lang.String, java.lang.String[])
	 */
	public long sadd(String key, String... members) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.sadd(key, members);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}
	
	private byte[] pop(String key, boolean isLeft){
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		byte[] values = null;
		try {
			if(isLeft){
				values = jedis.lpop(key.getBytes());
			}else{
				values = jedis.rpop(key.getBytes());
			}
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return values;
	}
	
	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#lpop(java.lang.String)
	 */
	public byte[] lpop(String key) {
		return pop(key, true);
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#rpoplpush(java.lang.String, java.lang.String)
	 */
	public byte[] rpoplpush(String scrKey, String dstKey) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		byte[] ret = null;
		try {
			Jedis j = jedis.getShard(scrKey + dstKey);
			ret = j.rpoplpush(scrKey.getBytes(), dstKey.getBytes());
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return ret;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#rpoplpushBean(java.lang.String, java.lang.String)
	 */
	@SuppressWarnings("unchecked")
	public <T extends Serializable> T rpoplpushBean(String scrKey, String dstKey) {
		return (T)HessianSerializerUtils.deSerialize(rpoplpush(scrKey, dstKey));
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#lrem(java.lang.String, byte[], long)
	 */
	public long lrem(String key, byte[] value, long count) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long ret = null;
		try {
			ret = jedis.lrem(key.getBytes(), count, value);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return ret == null ? 0 : ret.intValue();
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#lrem(java.lang.String, java.lang.String, long)
	 */
	public long lrem(String key, String value, long count) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long ret = null;
		try {
			ret = jedis.lrem(key, count, value);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return ret == null ? 0 : ret.intValue();
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#lpopBean(java.lang.String)
	 */
	@SuppressWarnings("unchecked")
	public <T extends Serializable> T lpopBean(String key) {
		return (T)HessianSerializerUtils.deSerialize(lpop(key));
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#rpop(java.lang.String)
	 */
	public byte[] rpop(String key) {
		return pop(key, false);
	}
	
	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#rpopBean(java.lang.String)
	 */
	@SuppressWarnings("unchecked")
	public <T extends Serializable> T rpopBean(String key) {
		return (T)HessianSerializerUtils.deSerialize(rpop(key));
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#lset(java.lang.String, byte[], long)
	 */
	public boolean lset(String key, byte[] value, long index) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		String ret = null;
		try {
			ret = jedis.lset(key.getBytes(), index, value);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return "OK".equals(ret) ? true : false;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#lset(java.lang.String, java.lang.String, long)
	 */
	public boolean lset(String key, String value, long index) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		String ret = null;
		try {
			ret = jedis.lset(key, index, value);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return "OK".equals(ret) ? true : false;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#lsetBean(java.lang.String, java.io.Serializable, long)
	 */
	public <T extends Serializable> boolean lsetBean(String key, T value,
			long index) {
		return lset(key, HessianSerializerUtils.serialize(value), index);
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#ltrim(java.lang.String, long, long)
	 */
	public boolean ltrim(String key, long start, long end) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		String value = null;
		try {
			value = jedis.ltrim(key, start, end);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return "OK".equals(value) ? true : false;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#zadd(java.lang.String, java.util.Map)
	 */
	public long zadd(byte[] key, Map<Double, byte[]> scoreMembers) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.zadd(key, scoreMembers);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#zadd(java.lang.String, java.util.Map)
	 */
	public long zadd(String key, Map<Double, String> scoreMembers) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.zadd(key, scoreMembers);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#zaddBean(java.lang.String, java.util.Map)
	 */
	public <T extends Serializable> long zaddBean(String key,
			Map<Double, T> scoreMembers) {
		if(scoreMembers != null && !scoreMembers.isEmpty()){
			Map<Double, byte[]> tmps = new HashMap<Double, byte[]>();
			Iterator<Entry<Double, T>> it = scoreMembers.entrySet().iterator();
			while(it.hasNext()){
				Entry<Double, T> entry = it.next();
				tmps.put(entry.getKey(), HessianSerializerUtils.serialize(entry.getValue()));
			}
			return zadd(key.getBytes(), tmps);
		}else{
			return 0;
		}
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#zaddPage(java.lang.String, java.util.List, int, int)
	 */
	public <T extends Serializable> long zaddPage(String key, List<T> items,
			int pageNo, int pageSize) {
		if(items != null && items.size() > 0){
			Map<Double, T> scoreMembers = new HashMap<Double, T>();
			for(int i=0; i< items.size();i++){
				double pos = (pageNo-1)*pageSize + i;
				scoreMembers.put(pos, items.get(i));
			}
			return zaddBean(key, scoreMembers);
		}
		return 0;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#zincrBy(java.lang.String, double, byte[])
	 */
	public double zincrBy(String key, double step, byte[] member) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Double value = null;
		try {
			value = jedis.zincrby(key.getBytes(), step, member);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#zincrBy(java.lang.String, double, java.lang.String)
	 */
	public double zincrBy(String key, double step, String member) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Double value = null;
		try {
			value = jedis.zincrby(key, step, member);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#zincrBy(java.lang.String, double, java.io.Serializable)
	 */
	public <T extends Serializable> double zincrBy(String key, double step,
			T member) {
		return zincrBy(key, step, HessianSerializerUtils.serialize(member));
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#zrem(java.lang.String, byte[][])
	 */
	public long zrem(String key, byte[]... members) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.zrem(key.getBytes(), members);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#zrem(java.lang.String, java.lang.String[])
	 */
	public long zrem(String key, String... members) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.zrem(key, members);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#zrem(java.lang.String, java.io.Serializable[])
	 */
	public <T extends Serializable> long zrem(String key, T... members) {
		if(members != null && members.length > 0){
			byte[][] tmps = new byte[members.length][];
			for(int i=0;i<members.length;i++){
				tmps[i] = HessianSerializerUtils.serialize(members[i]);
			}
			return zrem(key, tmps);
		}else{
			return 0;
		}
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#zremrangeByRank(java.lang.String, long, long)
	 */
	public long zremrangeByRank(String key, long start, long end) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.zremrangeByRank(key, start, end);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}

	/** (non-Javadoc)
	 * @see com.redis.api.DataWriter#zremrangeByScore(java.lang.String, double, double)
	 */
	public long zremrangeByScore(String key, double min, double max) {
		ShardedJedis jedis = this.shardedJedisPool.getResource();
		boolean flag = false;
		Long value = null;
		try {
			value = jedis.zremrangeByScore(key, min, max);
			flag = true;
		} catch (Exception e) {
			flag = false;
			e.printStackTrace();
		} finally {
			release(flag, jedis);
		}
		return value == null ? 0 : value;
	}
	
	
	
	
	
	

	

}
